package com.jixuian.rabbitmq.config;


import com.jixuian.rabbitmq.service.TransMessageService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Objects;

@Slf4j
@Configuration
public class RabbitConfig {

    @Value("${mq.host}")
    String host;
    @Value("${mq.port}")
    int port;
    @Value("${mq.username}")
    String username;
    @Value("${mq.password}")
    String password;
    @Value("${mq.vhost}")
    String vhost;

    @Autowired
    private TransMessageService transMessageService;

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(vhost);
        // 生产端确认, 使用CORRELATED是因为需要知道哪条消息, 因为确认是异步的
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        // 消息返回
        connectionFactory.setPublisherReturns(true);
        connectionFactory.createConnection();
        return connectionFactory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        // 可以自动扫描容器中声明的queue, exchange, binding并自动声明他们
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }

    @Bean
    public RabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setConcurrentConsumers(3);
        factory.setMaxConcurrentConsumers(10);
        // 手动ack
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        return factory;
    }

    @Bean
    public RabbitTemplate customRabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        // 消息返回
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returnedMessage) {
                log.error("消息发送失败, returnedMessage:{}", returnedMessage);
                transMessageService.messageSendReturn(returnedMessage.getMessage().getMessageProperties().getMessageId(),
                        returnedMessage.getExchange(),
                        returnedMessage.getRoutingKey(),
                        new String(returnedMessage.getMessage().getBody()));
            }
        });
        // 发送端确认
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                log.info("correlationData:{}, ack:{}, cause:{}", correlationData, ack, cause);
                if (Objects.nonNull(correlationData)) {
                    String correlationId = correlationData.getId();
                    transMessageService.messageSendSuccess(correlationId);
                } else {
                    log.error("消息投递失败, correlationData:{}, ack:{}, cause:{}", correlationData, ack, cause);
                }
            }
        });

        return rabbitTemplate;
    }
}

























